package kafka
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object TestSparkStreamingKafka{

  def main(args: Array[String]): Unit = {

    // 准备环境
    val spark: SparkSession = SparkSession.builder()
      .appName("TestSparkStreamingKafka")
      .master("local[*]")
      .getOrCreate()

    // 设置日志级别
    spark.sparkContext.setLogLevel("WARN")

    //  导入隐式转换
    import spark.implicits._

    // 读取数据流中的数据
    val kafkaDatas: DataFrame = spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", "dw-Huake05:9092,dw-Huake06:9092,dw-Huake07:9092")  // 设置kafka集群
      .option("subscribe", "test01")   // 设置需要读取的主题topic
      .load()  // 加载数据

    // kafkaDatas  内部数据是kafka数据(key,value)
    val kafkaDataString: Dataset[(String, String)] = kafkaDatas.selectExpr("CAST(key AS string)","CAST(value AS string)").as[(String,String)]

    // 处理，将数据按照空格切分
    val word: Dataset[String] = kafkaDataString.flatMap(x=>x._2.split(" "))

    // 利用DSL语句对数据进行wordcount
    val wordCount: Dataset[Row] = word.groupBy("value").count().sort($"count".desc)

    // 输出
    wordCount.writeStream.format("console")  // 输出方式,console表示控制台
      .outputMode("complete")  // 输出模式
      .start()  // 开启任务
      .awaitTermination()   // 等待程序停止

  }

}

